package com.eshipenet.shipnet.equipmentextbuilder
import java.lang

import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema
import org.apache.flink.streaming.util.serialization.KeyedSerializationSchema
import org.apache.kafka.clients.producer.ProducerRecord
import play.api.libs.json.JsValue

class JsonKafkaSerializer(topic: String) extends KafkaSerializationSchema[(String, JsValue)] {

  override def serialize(element: (String, JsValue), timestamp: lang.Long): ProducerRecord[Array[Byte], Array[Byte]] = {
    val key   = element._1.getBytes
    val value = element._2.toString().getBytes()
    new ProducerRecord(topic, key, value)
  }

}
